package site.teamo.biu.flink.tool.handler;

import cn.hutool.core.map.MapUtil;
import com.alibaba.fastjson2.JSON;
import com.alibaba.fastjson2.JSONObject;
import site.teamo.biu.flink.sdk.annotation.function.BFKeyedProcessFunction;
import site.teamo.biu.flink.sdk.annotation.function.BFProcessFunction;
import site.teamo.biu.flink.sdk.annotation.function.BFSinkFunction;
import site.teamo.biu.flink.sdk.annotation.function.BFSourceFunction;
import site.teamo.biu.flink.tool.BiuFlinkToolHandler;

import java.lang.annotation.Annotation;
import java.util.HashMap;
import java.util.Map;

/**
 * @author haocongshun
 * @date 2023/11/01 14:20:24
 */
public class VerifyFunctionPackageHandler implements BiuFlinkToolHandler {

    private static final Map<String, Class<? extends Annotation>> FUNCTION_TYPE_MAP = MapUtil.ofEntries(
            MapUtil.entry("source", BFSourceFunction.class),
            MapUtil.entry("process", BFProcessFunction.class),
            MapUtil.entry("keyed-process", BFKeyedProcessFunction.class),
            MapUtil.entry("sink", BFSinkFunction.class)
    );

    /**
     * @param in 要验证的class信息
     * @return 输出信息
     * 1: 类加载失败
     * 2: 算子实际类型与描述信息不符
     */
    @Override
    public String handle(String in) {
        JSONObject functionClassMap = JSON.parseObject(in);
        Map<String, String> result = new HashMap<>();
        for (String key : functionClassMap.keySet()) {
            Class<?> aClass;
            try {
                aClass = Class.forName(key);
            } catch (Exception e) {
                result.put(key, "类加载失败");
                continue;
            }
            Annotation annotation = aClass.getAnnotation(FUNCTION_TYPE_MAP.get(functionClassMap.getString(key)));
            if (annotation == null) {
                result.put(key, "算子实际类型与描述信息不符");
            }
        }
        return JSON.toJSONString(result);
    }
}
